package com.example.mq_sm.mqserver;

import com.example.mq_sm.common.Consumer;
import com.example.mq_sm.common.MQException;
import com.example.mq_sm.mqserver.core.*;
import com.example.mq_sm.mqserver.datacenter.DiskDataCenter;
import com.example.mq_sm.mqserver.datacenter.MemoryDataCenter;
import lombok.Data;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 通过这个类, 来表示虚拟主机
 * 每个虚拟主机下面都管理着自己的 交换机，队列，绑定，消息数据
 * 同时提供 api 供上层使用
 */
@Data
public class VirtualHost {
    private String virtualHostName;
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);

    // 设置锁对象
    private final Object exchangeLocker = new Object();
    private final Object queueLocker = new Object();

    public VirtualHost(String virtualHostName){
        this.virtualHostName = virtualHostName;

        // 对于 MemoryDataCenter 来说，不需要额外的初始化操作，只要将对象 new 出来即可
        // 对于 DiskDataCenter 来说，需要进行初始化操作，建立库表和初始数据
        diskDataCenter.init();

        // 对于还需要硬盘中的数据，需要恢复到内存中
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (MQException | IOException | ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复数据失败！");
        }
    }

    // 创建交换机
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete,
                                   Map<String, Object> arguments){
        exchangeName = virtualHostName + exchangeName;
        // 1. 先判断交换机是否已经被创建
        try{
            synchronized (exchangeLocker){
                Exchange exchangeExist = memoryDataCenter.getExchange(exchangeName);
                if (exchangeExist != null){
                    System.out.println("[VirtualHost] 交换机已经存在! exchangeName=" + exchangeName);
                    return true;
                }
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                if (durable){
                    diskDataCenter.insertExchange(exchange);
                }
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建完成! exchangeName=" + exchangeName);
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 交换机创建失败! exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    // 删除交换机
    public boolean exchangeDelete(String exchangeName){
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker){
                Exchange exchangeExist = memoryDataCenter.getExchange(exchangeName);
                if (exchangeExist == null){
                    throw new MQException("[VirtualHost] 交换机不存在无法删除!");
                }
                if (exchangeExist.isDurable()){
                    diskDataCenter.deleteExchange(exchangeName);
                }
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功! exchangeName=" + exchangeName);
            }

            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 交换机删除失败! exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    // 创建队列
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete,
                                Map<String, Object> arguments){
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker){
                MSGQueue queueExist = memoryDataCenter.getQueue(queueName);
                if (queueExist != null){
                    System.out.println("[VirtualHost] 队列已经存在! queueName=" + queueName);
                    return true;
                }
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                if (durable){
                    diskDataCenter.insertQueue(queue);
                }
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功! queueName=" + queueName);
            }

            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 队列创建失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 删除队列
    public boolean queueDelete(String queueName){
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker){
                MSGQueue queueExist = memoryDataCenter.getQueue(queueName);
                if (queueExist == null){
                    throw new MQException("[VirtualHost] 队列不存在! 无法删除! queueName=" + queueName);
                }
                if (queueExist.isDurable()){
                    diskDataCenter.deleteQueue(queueName);
                }
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 删除队列成功! queueName=" + queueName);
            }

            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 删除队列失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 创建绑定
    public boolean queueBind(String queueName, String exchangeName, String bindingKey){
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (queueLocker){
                synchronized (exchangeLocker){
                    Binding bindingExists = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (bindingExists != null){
                        throw new MQException("[VirtualHost] binding 已经存在! queueName=" + queueName
                                + ", exchangeName=" + exchangeName);
                    }
                    if (!router.checkBindingKey(bindingKey)){
                        throw new MQException("[VirtualHost] bindingKey 非法! bindingKey=" + bindingKey);
                    }
                    Binding binding = new Binding();
                    binding.setBindingKey(bindingKey);
                    binding.setQueueName(queueName);
                    binding.setExchangeName(exchangeName);
                    // 判断一下对应的交换机和队列是否存在
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if (queue == null){
                        throw new MQException("[VirtualHost] 队列不存在! queueName=" + queueName);
                    }

                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if (exchange == null){
                        throw new MQException("[VirtualHost] 交换机不存在! exchangeName=" + exchangeName);
                    }
                    // 需要两个都是可持久化的才能将这个绑定进行储存
                    if (queue.isDurable() && exchange.isDurable()){
                        diskDataCenter.insertBinding(binding);
                    }
                    memoryDataCenter.insertBinding(binding);
                }
            }
            System.out.println("[VirtualHost] 绑定创建成功! exchangeName=" + exchangeName
                    + ", queueName=" + queueName);
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 绑定创建失败! exchangeName=" + exchangeName
                    + ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }

    }
    // 删除绑定
    public boolean queueUnbind(String queueName, String exchangeName){
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (queueLocker){
                synchronized (exchangeLocker){
                    Binding toDelete = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (toDelete == null){
                        throw new MQException("[VirtualHost] 删除绑定失败! 绑定不存在! exchangeName=" + exchangeName + ", queueName=" + queueName);
                    }
                    diskDataCenter.deleteBinding(toDelete);
                    memoryDataCenter.deleteBinding(toDelete);
                    System.out.println("[VirtualHost] 删除绑定成功!");
                }
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 删除绑定失败!");
            e.printStackTrace();
            return false;
        }
    }

    // 发送消息到指定的交换机/队列中
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body){
        try {
            exchangeName = virtualHostName + exchangeName;
            // 判断 routingKey 是否合法
            if (!router.checkRoutingKey(routingKey)){
                throw new MQException("[VirtualHost] routingKey 非法! routingKey=" + routingKey);
            }

            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null){
                throw new MQException("[VirtualHost] 交换机不存在! exchangeName=" + exchangeName);
            }
            // 判断交换机的类型
            if (exchange.getType() == ExchangeType.DIRECT){
                // 按照直接交换机的方式来进行消息的转发
                // 以 routingKey 作为队列的名字，直接把消息写入指定的消息队列当中
                // 此时，可以无视绑定关系
                String queueName = virtualHostName + routingKey;
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null){
                    throw new MQException("[VirtualHost] 队列不存在! queueName=" + queueName);
                }
                // 队列存在，直接转发消息
                sendMessage(queue, message);
                return true;
            }else{
                // 按照 fanout 和 topic 的方式来进行转发
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                for (Map.Entry<String, Binding> entry : bindingsMap.entrySet()){
                    // 获取一下对应的队列，看一下队列是否存在
                    Binding binding = entry.getValue();
                    String queueName = virtualHostName + binding.getQueueName();
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
//                    String queueName = virtualHostName + entry.getKey();
//                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if (queue == null){
                        //throw new MQException("");
                        // 其他队列可能ok，不能因为一个小问题影响全部
                        // 所以在这里不抛异常，而是选择打印一个日志
                        System.out.println("[VirtualHost] basicPublish 发送消息时, 发现队列不存在! queueName=" + binding.getQueueName());
                        continue;
                    }
                    Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                    // fanout 所有绑定的队列都需要进行转发的
                    // topic 需要进行判定 routingKey 和 bindingKey 进行判断
                    if (!router.route(exchange.getType(), binding, message)){
                        continue;
                    }
                    sendMessage(queue, message);
                }
                return true;
            }
        }catch (Exception e){
            System.out.println("[VirtualHost] 消息发送失败!");
            e.printStackTrace();
            return false;
        }
    }

    public void sendMessage(MSGQueue queue, Message message) throws MQException, IOException, InterruptedException {
        // 发送消息就是将消息写入硬盘和内存上面
        int deliverMode = message.getDeliverMode();
        if (deliverMode == 2){
            diskDataCenter.sendMessage(queue, message);
        }
        memoryDataCenter.sendMessage(queue, message);

        // TODO 通知消费者可以消费消息了
        consumerManager.notifyConsume(queue.getName());

    }

    // 订阅消息
    // 添加一个队列的订阅者，当队列收到消息之后，就要把消息推送给对应的订阅者
    // consumerTag 消费者的身份标识
    // autoAck 消息被消费完成后，应答的方式 为 true 为自动应答，false 为手动应答
    // consumer 是一个回调函数 此处类型设定成函数式接口，这样后续调用 basicConsume 并且传实参的时候就可以写成 lambda 表达式的样子
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer){
        queueName = virtualHostName + queueName;
        try {
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
            System.out.println("[VirtualHost] basicConsume 成功! queueName=" + queueName);
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] basicConsume 失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean basicAck(String queueName, String messageId){
        queueName = virtualHostName + queueName;
        try{
            // 1. 获取到消息和队列
            Message message = memoryDataCenter.getMessage(messageId);
            if (message == null){
                throw new MQException("[VirtualHost] 要确认的消息不存在! messageId=" + messageId);
            }
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null){
                throw new MQException("[VirtualHost] 要确认的队列不存在! queueName=" + queueName);
            }
            // 现在开始删除数据
            if (message.getDeliverMode() == 2){
                diskDataCenter.deleteMessage(queue, message);
            }
            memoryDataCenter.removeMessage(messageId);
            memoryDataCenter.removeMessageWaitAck(queueName, messageId);
            System.out.println("[VirtualHost] basicAck 成功! 消息被成功确认! queueName=" + queueName
                    + ", messageId=" + messageId);
            return true;
        } catch (Exception e){
            System.out.println("[VirtualHost] basicAck 失败! 消息确认失败! queueName=" + queueName
                    + ", messageId=" + messageId);
            e.printStackTrace();
            return false;
        }
    }
}





















